package shareVal

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.{After, Before, Test}

class BroadcastTest {


	val conf: SparkConf = new SparkConf()
	var sc: SparkContext = null
	var outpath: String = "out"

	import util.MyPredef._

	@Before
	def init() {

		conf.setAppName("WordCount")
		conf.setMaster("local")
		sc = new SparkContext(conf)

		outpath.delete
	}

	@After
	def after() {
		sc.stop()
	}


	/**
	 * 广播变量 : 用于代替Join 调优
	 */
	@Test
	def accumulaterTest(): Unit = {

		val broadcast: Broadcast[List[(Int, Int)]] = sc.broadcast(List((1, 1), (2, 2), (3, 3)))

		val ins: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (2, "b"), (3, "c")), 2)

		val res: RDD[(Int, (String, Any))] = ins.map {
			case (key, value) => {
				var v2: Any = null
				for (t <- broadcast.value if key == t._1) {
					v2 = t._2
				}

				(key, (value, v2))
			}

		}
		res.foreachPartition(x => println(x.toArray.mkString(", ")))
	}
}
